package com.shu.imp;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;

public class WeightUserDay {
	
	private static Logger log=Logger.getLogger(WeightUserDay.class);
	public static void move(RestHighLevelClient esClient ,RestHighLevelClient client, String indexName) throws Exception {
		try {
			
			List<String> list=new ArrayList<String>();
			
			
			String pathname = "D:\\cc1.txt"; // 绝对路径或相对路径都可以，这里是绝对路径，写入文件时演示相对路径  
	        File filename = new File(pathname); // 要读取以上路径的input。txt文件  
	        InputStreamReader reader = new InputStreamReader(  
	                new FileInputStream(filename)); // 建立一个输入流对象reader  
	        BufferedReader br = new BufferedReader(reader); // 建立一个对象，它把文件内容转成计算机能读懂的语言  
	        String line = "";  
	        line = br.readLine();  
	        while (line!= null) {  
	            line = br.readLine(); // 一次读入一行数据  
	            System.out.println(line);
	            if(line!=null && !line.equals("")) {
	            	list.add(line.replace(" ", ""));
	            }
	        }
	        
	        
			SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
			searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("official_accounts_appid", list))
																.must(QueryBuilders.rangeQuery("create_time").lte("20180123235959")));
			searchSourceBuilder.sort("create_time", SortOrder.ASC);
			String[] indes = {"weighthistory_day2018"};
			SearchRequest searchRequest = new SearchRequest(indes,searchSourceBuilder);
			SearchResponse response = esClient.search(searchRequest);
			SearchHit[] hits = response.getHits().getHits();
			int count = (int)(response.getHits().getTotalHits());
			System.out.println("========================================>"+count);
				
			Page page=new Page(count, 5000);
			for (int i = 1; i <= page.getTotalPages(); i++) {
				System.out.println(page.getStart(i));
				query(page.getStart(i), page.getPageSize(),client,indexName,esClient,list);
			}
			
			
		} catch (Exception e) {
			log.error(e.getMessage(),e);
			throw e;
		}
	}
	
	public static void query(int start,int size, RestHighLevelClient client,String indexName,RestHighLevelClient esClient,List<String> list) throws Exception {
		try {
			SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
			searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("official_accounts_appid", list))
																.must(QueryBuilders.rangeQuery("create_time").lte("20180123235959")));
			searchSourceBuilder.from(start);
			searchSourceBuilder.size(size);
			searchSourceBuilder.sort("create_time", SortOrder.ASC);
			String[] indes = {"weighthistory_day2018"};
			SearchRequest searchRequest = new SearchRequest(indes,searchSourceBuilder);
			SearchResponse response = esClient.search(searchRequest);
			SearchHit[] hits = response.getHits().getHits();
			
			BulkRequest request = new BulkRequest();
			for (SearchHit hit:hits) {
				Map<String,Object> map = hit.getSourceAsMap();
				request.add(new IndexRequest(indexName+1, "data",hit.getId()).source(map));
			}
			request.timeout(TimeValue.timeValueMinutes(10000000));
			if(request.numberOfActions()==0) {
				return;
			}
			BulkResponse bulk = client.bulk(request);
			if(bulk.hasFailures()){
				bulk = client.bulk(request);
			}
			
			if(bulk.hasFailures()){
				bulk = client.bulk(request);
				if(bulk.hasFailures()){
					log.error(bulk.buildFailureMessage());
					log.info("====================================ES data has submited fail! "+start +":");
				}
			}else {
				log.info("====================================ES data has submited sucess! "+start +":");
			}
		}catch (Exception e) {
			log.error(e.getMessage(),e);
			throw e;
		}
		
	}
	
	public static List<String> findDates(Date dBegin, Date dEnd)  
	 { 
      SimpleDateFormat sd=new SimpleDateFormat("yyyyMMdd");	
	  List<String> lDate = new ArrayList<String>();  
	  lDate.add(sd.format(dBegin).toString());  
	  Calendar calBegin = Calendar.getInstance();  
	  // 使用给定的 Date 设置此 Calendar 的时间  
	  calBegin.setTime(dBegin);  
	  Calendar calEnd = Calendar.getInstance();  
	  // 使用给定的 Date 设置此 Calendar 的时间  
	  calEnd.setTime(dEnd);  
	  // 测试此日期是否在指定日期之后  
	  while (dEnd.after(calBegin.getTime()))  
	  {  
	   // 根据日历的规则，为给定的日历字段添加或减去指定的时间量  
	   calBegin.add(Calendar.DAY_OF_MONTH, 1);  
	   lDate.add(sd.format(calBegin.getTime()).toString());  
	  }  
	  return lDate;  
	 }
	
	public static RestHighLevelClient getClient(String ip,Integer port) {
		List<HttpHost> list = new ArrayList<HttpHost>();
		list.add(new HttpHost(ip, port, "http"));
		HttpHost[] arr = new HttpHost[list.size()];
		arr = list.toArray(arr);
		
		RestClientBuilder builder = RestClient.builder(arr)
				.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
					public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
						requestConfigBuilder.setConnectTimeout(50000);
						requestConfigBuilder.setSocketTimeout(50000);
						requestConfigBuilder.setConnectionRequestTimeout(50000);
						return requestConfigBuilder;
					}

				}).setMaxRetryTimeoutMillis(20 * 60 * 1000);

		return new RestHighLevelClient(builder);
	}
	
	public static void main(String[] args) throws Exception {
		move(getClient("127.0.0.1",9200), getClient("127.0.0.1",9202), "weighthistory_day");
	}
}
